#include "collectives.h"
#include "type.h"
#include "transport.h"
#include "net.h"
#include "group.h"


flagcxResult_t flagcxHeteroSend(const void* sendbuff, size_t count, flagcxDataType_t datatype, int peer,
                flagcxHeteroComm_t comm, flagcxStream_t stream){
    flagcxHeteroGroupStart();
    int channelId = 0;
    if(comm->channels[channelId].peers[peer]->send[0].connected == 0){
        comm->connectSend[peer] |= (1UL<<channelId);
        flagcxGroupCommPreconnect(comm);
    }
    struct flagcxTaskP2p* p2p;
    struct flagcxTasks *tasks = &comm->tasks;
    FLAGCXCHECK(flagcxCalloc(&p2p, 1));
    p2p->buff = (void *)sendbuff;
    p2p->bytes = count*getFlagcxDataTypeSize(datatype);
    p2p->chunk = 0;
    p2p->stream = stream;
    if(flagcxIntruQueueEmpty(&tasks->peers[peer].sendQueue)) tasks->p2pOrder[tasks->p2pOrderSteps++]=peer;
    flagcxIntruQueueEnqueue(&tasks->peers[peer].sendQueue, p2p);

    flagcxGroupCommJoin(comm);
    flagcxHeteroGroupEnd();
    return flagcxSuccess;
}


flagcxResult_t flagcxHeteroRecv(void* recvbuff, size_t count, flagcxDataType_t datatype, int peer,
    flagcxHeteroComm_t comm, flagcxStream_t stream) {
    flagcxHeteroGroupStart();
    int channelId = 0;

    if(comm->channels[channelId].peers[peer]->recv[0].connected == 0){
        comm->connectRecv[peer] |= (1UL<<channelId);
        flagcxGroupCommPreconnect(comm);
    }
    struct flagcxTaskP2p* p2p;
    struct flagcxTasks *tasks = &comm->tasks;
    FLAGCXCHECK(flagcxCalloc(&p2p, 1));
    p2p->buff = (void *)recvbuff;
    p2p->bytes = count*getFlagcxDataTypeSize(datatype);
    p2p->chunk = 0;
    p2p->stream = stream;
    if(flagcxIntruQueueEmpty(&tasks->peers[peer].recvQueue)) tasks->p2pOrder[tasks->p2pOrderSteps++]=peer;
    flagcxIntruQueueEnqueue(&tasks->peers[peer].recvQueue, p2p);

    flagcxGroupCommJoin(comm);
    flagcxHeteroGroupEnd();
    return flagcxSuccess;
}